package com.itbaima.csmall.cart.webapi.kafka;

import com.google.gson.Gson;
import com.itbaima.csmall.commons.pojo.cart.entity.Cart;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * <p>Consumer</p>
 *
 * @author Chen
 * @since 2023/3/8 10:55:55
 */
@Slf4j
@Component
public class Consumer {

    @KafkaListener(topics = Producer.TOPIC_KEY)
    public void processMessage(ConsumerRecord<String,String> record) {
        String json = record.value();
        Gson gson = new Gson();
        Cart cart = gson.fromJson(json, Cart.class);
        log.info("<==Kafka接收到消息：{}", cart);
    }

    @RabbitListener(queues = Producer.TOPIC_KEY)
    public void receive(@Payload String body, @Headers Map<String, Object> headers) {
        log.info("<==RabbitMQ接收到消息：{}, {}", body, headers);
    }

}
